package com.gzsxy.rabbitmq.producer.service.impl;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.gzsxy.rabbitmq.producer.service.RabbitMQService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

/**
 * <p>
 * RabbitMQ生产端常用方法整合
 * </p>
 *
 * @author xiaolong
 **/
@Slf4j
@Service
public class RabbitMQServiceImpl implements RabbitMQService {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private ObjectMapper objectMapper;

    @Override
    public void sendMessageByWork(String queue, Object msg) {
        //将对象转为字节数组
        byte[] writeValueAsBytes = null;
        try {
            writeValueAsBytes = objectMapper.writeValueAsBytes(msg);
            ////主要是将消息存储在body中的，将消息的deLiveryMode设置为持久与永久的
            Message message = MessageBuilder.withBody(writeValueAsBytes).setDeliveryMode(MessageDeliveryMode.PERSISTENT).build();
            message.getMessageProperties().setContentType(MessageProperties.CONTENT_TYPE_JSON);
            rabbitTemplate.convertAndSend("", queue, msg);
        } catch (JsonProcessingException e) {
            log.error("发送消息失败", e);
            throw new RuntimeException("发送消息失败，请检查rabbitmq服务");
        }
    }

    @Override
    public void sendMessageByWork(String queue, Object msg, String ackId) {
        //将对象转为字节数组
        byte[] writeValueAsBytes = null;
        try {
            writeValueAsBytes = objectMapper.writeValueAsBytes(msg);
            ////主要是将消息存储在body中的，将消息的deLiveryMode设置为持久与永久的
            Message message = MessageBuilder.withBody(writeValueAsBytes).setDeliveryMode(MessageDeliveryMode.PERSISTENT).build();
            message.getMessageProperties().setContentType(MessageProperties.CONTENT_TYPE_JSON);
            rabbitTemplate.convertAndSend("", queue, msg, new CorrelationData(ackId));
        } catch (JsonProcessingException e) {
            log.error("发送消息失败", e);
            throw new RuntimeException("发送消息失败，请检查rabbitmq服务");
        }
    }

    @Override
    public void sendMessageByExchange(String exchange, String routingKey, Object msg) {
        //将对象转为字节数组
        byte[] writeValueAsBytes = null;
        try {
            writeValueAsBytes = objectMapper.writeValueAsBytes(msg);
            ////主要是将消息存储在body中的，将消息的deLiveryMode设置为持久与永久的
            Message message = MessageBuilder.withBody(writeValueAsBytes).setDeliveryMode(MessageDeliveryMode.PERSISTENT).build();
            message.getMessageProperties().setContentType(MessageProperties.CONTENT_TYPE_JSON);
            rabbitTemplate.convertAndSend(exchange, routingKey == null ? "" : routingKey, message);
        } catch (JsonProcessingException e) {
            log.error("发送消息失败", e);
            throw new RuntimeException("发送消息失败，请检查rabbitmq服务");
        }

    }

    @Override
    public void sendMessageByExchange(String exchange, String routingKey, Object msg, String ackId) {
        //将对象转为字节数组
        byte[] writeValueAsBytes = null;
        try {
            writeValueAsBytes = objectMapper.writeValueAsBytes(msg);
            ////主要是将消息存储在body中的，将消息的deLiveryMode设置为持久与永久的
            Message message = MessageBuilder.withBody(writeValueAsBytes).setDeliveryMode(MessageDeliveryMode.PERSISTENT).build();
            message.getMessageProperties().setContentType(MessageProperties.CONTENT_TYPE_JSON);
            rabbitTemplate.convertAndSend(exchange, routingKey == null ? "" : routingKey, message, new CorrelationData(ackId));
        }   catch (JsonProcessingException e) {
              log.error("发送消息失败", e);
              throw new RuntimeException("发送消息失败，请检查rabbitmq服务");
        }

    }

    @Override
    public void sendMessageByDelayedeExchange(String exchange, String routingKey, Object msg, Integer delayTime, String ackId) {
        //将对象转为字节数组
        byte[] writeValueAsBytes = null;
        try {
            writeValueAsBytes = objectMapper.writeValueAsBytes(msg);
            ////主要是将消息存储在body中的，将消息的deLiveryMode设置为持久与永久的
            Message message = MessageBuilder.withBody(writeValueAsBytes).setDeliveryMode(MessageDeliveryMode.PERSISTENT).build();
            message.getMessageProperties().setContentType(MessageProperties.CONTENT_TYPE_JSON);
            rabbitTemplate.convertAndSend(exchange, routingKey == null ? "" : routingKey, message,  correlationData -> {
                correlationData.getMessageProperties().setDelay(delayTime);
                return correlationData;
            },new CorrelationData(ackId));
        }   catch (JsonProcessingException e) {
            log.error("发送消息失败", e);
            throw new RuntimeException("发送消息失败，请检查rabbitmq服务");
        }
    }
}
